package main

import (
	"fmt"
	"log"
	"time"
)

func main() {
	fmt.Println("DataFlare Simple Example")
	fmt.Println("------------------------")
	fmt.Println("This is a simple example of DataFlare in action.")
	fmt.Println("It demonstrates the core concepts of DataFlare:")
	fmt.Println("1. Data sources")
	fmt.Println("2. Data processors")
	fmt.Println("3. Data sinks")
	fmt.Println("4. Data flow")
	fmt.Println()

	// Create a simple data source
	source := NewMemorySource([]map[string]interface{}{
		{"id": 1, "name": "Alice", "age": 30},
		{"id": 2, "name": "Bob", "age": 25},
		{"id": 3, "name": "Charlie", "age": 35},
	})

	// Create a simple data processor
	processor := NewAgeProcessor()

	// Create a simple data sink
	sink := NewConsoleSink()

	// Connect the components
	source.SetNext(processor)
	processor.SetNext(sink)

	// Start the data flow
	fmt.Println("Starting data flow...")
	source.Start()

	// Wait for the data flow to complete
	time.Sleep(1 * time.Second)
	fmt.Println("Data flow completed!")
}

// Component is a data flow component
type Component interface {
	Process(data map[string]interface{}) error
	SetNext(next Component)
}

// BaseComponent is a base implementation of Component
type BaseComponent struct {
	next Component
}

// SetNext sets the next component in the chain
func (c *BaseComponent) SetNext(next Component) {
	c.next = next
}

// Forward forwards data to the next component
func (c *BaseComponent) Forward(data map[string]interface{}) error {
	if c.next != nil {
		return c.next.Process(data)
	}
	return nil
}

// MemorySource is a data source that reads from memory
type MemorySource struct {
	BaseComponent
	data []map[string]interface{}
}

// NewMemorySource creates a new memory source
func NewMemorySource(data []map[string]interface{}) *MemorySource {
	return &MemorySource{
		data: data,
	}
}

// Start starts the data source
func (s *MemorySource) Start() {
	for _, data := range s.data {
		err := s.Process(data)
		if err != nil {
			log.Printf("Error processing data: %v", err)
		}
	}
}

// Process processes data
func (s *MemorySource) Process(data map[string]interface{}) error {
	return s.Forward(data)
}

// AgeProcessor is a data processor that processes age
type AgeProcessor struct {
	BaseComponent
}

// NewAgeProcessor creates a new age processor
func NewAgeProcessor() *AgeProcessor {
	return &AgeProcessor{}
}

// Process processes data
func (p *AgeProcessor) Process(data map[string]interface{}) error {
	// Get age
	age, ok := data["age"].(int)
	if !ok {
		return fmt.Errorf("age is not an integer")
	}

	// Add age category
	if age < 30 {
		data["category"] = "young"
	} else {
		data["category"] = "adult"
	}

	return p.Forward(data)
}

// ConsoleSink is a data sink that writes to the console
type ConsoleSink struct {
	BaseComponent
}

// NewConsoleSink creates a new console sink
func NewConsoleSink() *ConsoleSink {
	return &ConsoleSink{}
}

// Process processes data
func (s *ConsoleSink) Process(data map[string]interface{}) error {
	fmt.Printf("Data: %v\n", data)
	return nil
}
